/*
异步工作器的设计
*/

#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__
#include "buffer.hpp"
#include <iostream>
#include <condition_variable>
#include <functional>
#include <thread>
#include <memory>
#include <atomic>
#include <mutex>
#include <condition_variable>

namespace clog
{
    enum class AsyncType
    {
        ASYNC_SAVE = 0,
        ASYNC_UNSAVE
    };
    class AsyncLooper
    {
        using Functor = std::function<void(Buffer &)>;

    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor &cb, AsyncType looper_type = AsyncType::ASYNC_SAVE)
            : _stop(false), _thread(std::thread(&AsyncLooper::threadEntry, this)), _cb(cb), _looper_type(looper_type)
        {
        }
        void push(const char *data, size_t len)
        {
            // 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            // 判断是否有足够的空间进行写入，是否需要阻塞，缓冲区满了就不能再写入
            if (_looper_type == AsyncType::ASYNC_SAVE)
            {
                // 阻塞
                _cond_proc.wait(lock, [&]()
                                { return len <= _pro_buf.writeAbleSize(); });
            }
            // 写入条件满足
            _pro_buf.push(data, len);
            // 唤醒异步工作线程
            _cond_con.notify_all();
        }
        void stop()
        {
            _stop = true;
            // 唤醒所有异步工作线程
            _cond_con.notify_all();
            _thread.join();
        }
        ~AsyncLooper()
        {
            stop();
        }

    private:
        void threadEntry()
        {
            while (1)
            {
                // 限定加锁的生命周期，因为在读取完数据后，处理数据的过程和放入数据的过程是可以并行的！
                {
                    // 加锁
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 异步工作器已经退出并且输入缓冲区数据读完了，就退出结束交换读取操作
                    if (_stop && _pro_buf.empty())
                        break;
                    // 如果输入缓冲区中没有数据，就无法交换缓冲区，就会被阻塞
                    _cond_con.wait(lock, [&]()
                                   { return _stop || !_pro_buf.empty(); });
                    // 走到这里说明输入缓冲区中有数据，可以进行交换读取
                    _pro_buf.swap(_con_buf);
                }
                // 交换完后就可以继续向输入缓冲区中push数据了
                if (_looper_type == AsyncType::ASYNC_SAVE)
                {
                    // 唤醒业务线程
                    _cond_proc.notify_all();
                }
                // 处理数据--调用回调函数
                _cb(_con_buf);
                // 初始化处理缓冲区
                _con_buf.reset();
            }
        }

    private:
        AsyncType _looper_type;
        Buffer _con_buf;         // 处理缓冲区
        Buffer _pro_buf;         // 输入缓冲区
        std::atomic<bool> _stop; // 异步工作器停止标志
        std::mutex _mutex;       // 互斥锁
        std::condition_variable _cond_con;
        std::condition_variable _cond_proc;
        std::thread _thread; // 处理数据的异步工作线程
    private:
        Functor _cb; // 处理数据的回调函数
    };

}

#endif